package GEMonitor;

import java.util.concurrent.*;

import io.netty.channel.Channel;

public class GEChannelMgr {
    private static GEChannelMgr _ChannelMgr = null;

    private ConcurrentHashMap<String, Channel> _channels = new ConcurrentHashMap<String, Channel>();
    private ConcurrentLinkedQueue<GEMsg> _reports = new ConcurrentLinkedQueue<GEMsg>();
    private MysqlHelper db = new MysqlHelper();

    private GEChannelMgr() {

    }

    public static GEChannelMgr getInstance() {
        if (_ChannelMgr == null) {
            _ChannelMgr = new GEChannelMgr();
        }
        return _ChannelMgr;
    }

    public void addChannel(String systemID, Channel ch) {
        _channels.put(systemID, ch);
    }

    public void delChannel(String systemID) {
        _channels.remove(systemID);
    }

    public boolean hasChannel(String systemID)
    {
        return _channels.get(systemID) != null;
    }

    public void addReport(GEMsg msg) {
        _reports.add(msg);
    }

    public void addReport(GEMsg msg, Channel ch) 
    {
            _channels.put(msg.systemID, ch);
//            _reports.add(msg);

    }

    public GEMsg pollReport() {
        return _reports.poll();
    }

    public boolean connect() {
        return db.connect();
    }

    public void processReport() {
        while (true) {  
            try {
                //Thread.sleep(1);
                if(_reports == null || _reports.isEmpty())
                {
                    continue;
                }

                GEMsg msg = _reports.poll();

                if (msg != null) {
//                    GEKafkaProducer.send(msg.systemID,new String(msg.content));
                    db.addReport(msg);
                }
            } catch (Exception e) {
                //e.printStackTrace();
                MyLogger.Error("processReport error: " + e.getMessage());
            }
        } 
    }

    public void processCmd()
    {
        while(true)
        {
            try {
                Thread.sleep(1000);
//                MyLogger.Info("processCmd ing... ");

                GEMsg msg = db.getCommand();
                if(msg == null)
                {
                    continue;
                }

//                MyLogger.Info("get one command: " + msg.systemID);

                Channel ch = _channels.get(msg.systemID);
                if(ch != null)
                {
                    MyLogger.Info("got channel by:" + msg.systemID);

                    if(ch.isActive() && ch.isWritable())
                    {                       
                        MyLogger.Info("channel is active, write and flush:" + msg.systemID);
                        ch.writeAndFlush(msg);                       
                    }
                    else
                    {
                        MyLogger.Info("channel is inactive, delete and continue:" + msg.systemID);
                        //ch.close();
                        delChannel(msg.systemID);
                    }                   
                }
                else
                {
                    MyLogger.Warn("not get channel by:" + msg.systemID);
                }
            } catch (Exception e) {
                //e.printStackTrace();
                MyLogger.Error("processCmd error: " + e.getMessage());
            }
        }
    }

    public void start()
    {
        connect();

        ExecutorService pool = Executors.newFixedThreadPool(1);
        try {
            pool.submit(this::processCmd);
        } catch (Exception e){
            MyLogger.Error("sendCmd error: " + e.getMessage());
        }


//        new Thread(new Runnable() {
//            @Override
//            public void run() {
//                processCmd();
//            }
//        }).start();

//        new Thread(new Runnable() {
//            @Override
//            public void run() {
//                processReport();
//            }
//        }).start();
    }
}